package com.czxy

import com.czxy.bean.{HBaseMeta, TagRule}

import java.util.Properties

import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

/**
  * @author Hxm
  * @date 2020/6/10 12:01
  */
object PoliticalFace {
  def main(args: Array[String]): Unit = {
    //1  创建SparkSQL对象
    //   用于读取mysql ， hbase等数据
    val spark: SparkSession = SparkSession.builder().appName("politicalFaceTag").master("local[*]").getOrCreate()

    //2  连接mysql 数据库
    //url: String, table: String, properties: Properties
    var url = "jdbc:mysql://bd001:3306/tags_new?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&user=root&password=123456"
    var table = "tbl_basic_tag" //mysql数据表的表名
    var properties: Properties = new Properties

    //连接mysql
    val mysqlConn: DataFrame = spark.read.jdbc(url, table, properties)


    //引入隐式转换
    import spark.implicits._
    //引入java 和scala相互转换
    import scala.collection.JavaConverters._
    //引入sparkSQL的内置函数
    import org.apache.spark.sql.functions._

    //3  读取mysql数据库中方的四级标签
    //   为读取hbase数据做准备
    //inType=HBase##zkHosts=192.168.10.20##zkPort=2181##hbaseTable=tbl_users##family=detail##selectFields=id,politicalFace
    val fourDS: Dataset[Row] = mysqlConn.select("rule").where("id = 64")
    val fourMap: Map[String, String] = fourDS.map(row => {
      //使用 ## 切分 再用 =  切分
      row.getAs("rule").toString.split("##")
        .map(kv => {
          val arr = kv.split("=")
          (arr(0), arr(1))
        })
    }).collectAsList().get(0).toMap

    //将map转换 HBaseMeta成样例类
    var hbaseMeta: HBaseMeta = getHBaseMeta(fourMap)

    //打印数据
    println(hbaseMeta)

    //4  读取mysql数据库中方的五级标签
    //   匹配职业
    val fiveDS: Dataset[Row] = mysqlConn.select("id", "rule").where("pid = 64")
    //封装成 TagRule样例类
    val fiveTagRule: List[TagRule] = fiveDS.map(row => {
      //获取数据
      val id: Int = row.getAs("id").toString.toInt
      val rule: String = row.getAs("rule").toString
      //封装
      TagRule(id, rule)
    }).collectAsList() //将DataSet传换成util.List[TagRule]   这个类型遍历时无法获取id,rule数据
      .asScala.toList //将util.List转换成list   需要隐式转换    import scala.collection.JavaConverters._

    //打印数据

    for (a <- fiveTagRule) {
      println(a.id + "   " + a.rule)
    }
    //5  根据mysql数据中的四级标签的规则   读取hbase 数据
    val HBaseDatas: DataFrame = spark.read.format("com.czxy.tools.HBaseDataSource")
      .option(HBaseMeta.ZKHOSTS, hbaseMeta.zkHosts)
      .option(HBaseMeta.ZKPORT, hbaseMeta.zkPort)
      .option(HBaseMeta.HBASETABLE, hbaseMeta.hbaseTable)
      .option(HBaseMeta.FAMILY, hbaseMeta.family)
      .option(HBaseMeta.SELECTFIELDS, hbaseMeta.selectFields)
      .load()
    //打印数据

    HBaseDatas.show()

    //+---+-------------+
    //| id|politicalFace|
    //+---+-------------+
    //|  1|            1|
    //| 10|            2|

    //6  标签匹配
    //    根据五级标签数据和hbase数据进行标签匹配  的到最终的标签
    //
    val getTags = udf((rule: String) => {
      //默认的tagId = 0
      var tagsId = 0
      for (tagRule <- fiveTagRule) {
        if (tagRule.rule == rule) {
          tagsId = tagRule.id
        }
      }
      tagsId
    })

    val JobTags = HBaseDatas.select('id as ("userId"), getTags('politicalFace).as("tagsId"))
    //打印数据

    JobTags.show()

    //+------+------+
    //|userId|tagsId|
    //+------+------+
    //|     1|    65|
    //|    10|    66|

    //新数据（job）的结构  userId   tagsId
    //老数据（politicalFace）的结构  userId   tagsId
    //使用join 将两个数据的tagsId合并到一起

    var getAllTagas = udf((oldTagsId: String, newTagsId: String) => {
      if (oldTagsId == "") {
        newTagsId
      } else if (newTagsId == "") {
        oldTagsId
      } else if (oldTagsId == "" && newTagsId == "") {
        ""
      } else {
        //拼接历史数据和新数据（可能有重复的数据）
        val alltags = oldTagsId + "," + newTagsId
        //对重复数据区中去重
        alltags.split(",").distinct
          //使用逗号分隔，返回字符串类型。
          .mkString(",")
      }
    })


    //7、解决数据覆盖的问题【职业标签会覆盖前面的所有标签】
    //   读取test，追加标签后覆盖写入
    // 读取test内的历史标签数据，追加新计算出来的标签到历史数据，最后覆盖写入hbase
    //A    读取test内的历史标签数据【不是职业标签，是其他的，已经计算出来的标签】
    val oldTags: DataFrame = spark.read.format("com.czxy.tools.HBaseDataSource")
      .option(HBaseMeta.ZKHOSTS, hbaseMeta.zkHosts)
      .option(HBaseMeta.ZKPORT, hbaseMeta.zkPort)
      .option(HBaseMeta.HBASETABLE, "test")
      .option(HBaseMeta.FAMILY, "detail")
      .option(HBaseMeta.SELECTFIELDS, "userId,tagsId")
      .load()
    //打印数据

    oldTags.show()

    //    +------+----------+
    //    |userId|    tagsId|
    //    +------+----------+
    //    |     1|6,10,63,65|
    //      |    10|6,12,63,66|


    //B    追加新计算出来的标签到历史数据
    //新表join新表，条件是两个表的userId相等
    val joinTagas: DataFrame = oldTags.join(JobTags, oldTags("userId") === JobTags("userId"))
    joinTagas.show()

    //|userId|    tagsId|userId|tagsId|
    //+------+----------+------+------+
    //|   296|5,13,63,65|   296|    65|
    //|   467|6,13,63,65|   467|    65|


    val allTags: DataFrame = joinTagas.select(
      //处理第一个字段    两个表中的多个userId字段，只读取一个
      when((oldTags.col("userId").isNotNull), (oldTags.col("userId")))
        .when((JobTags.col("userId").isNotNull), (JobTags.col("userId")))
        .as("userId"),

      //处理第二个字段  将两个字段个合并一起
      //自定义函数用于做数据的凭借
      getAllTagas(oldTags.col("tagsId"), JobTags.col("tagsId")).as("tagsId")
    )
    //打印数据

    allTags.show()
    //|userId|    tagsId|
    //+------+----------+
    //|   296|5,13,63,65|
    //|   467|6,13,63,65|


    //C    最后覆盖写入hbase

        //8 将最终的标签写入hbase
        allTags.write.format("com.czxy.tools.HBaseDataSource")
          .option(HBaseMeta.ZKHOSTS, hbaseMeta.zkHosts)
          //HBaseMeta.ZKPORT=>"zkPort"
          //hbaseMeta.zkPort=>2181
          .option(HBaseMeta.ZKPORT, hbaseMeta.zkPort)

          .option(HBaseMeta.HBASETABLE, "test")
          .option(HBaseMeta.FAMILY, "detail")
          .option(HBaseMeta.SELECTFIELDS, "userId,tagsId")
          .save()

  }

  def getHBaseMeta(fourMap: Map[String, String]): HBaseMeta = {
    HBaseMeta(
      fourMap.getOrElse(HBaseMeta.INTYPE, ""),
      fourMap.getOrElse(HBaseMeta.ZKHOSTS, ""),
      fourMap.getOrElse(HBaseMeta.ZKPORT, ""),
      fourMap.getOrElse(HBaseMeta.HBASETABLE, ""),
      fourMap.getOrElse(HBaseMeta.FAMILY, ""),
      fourMap.getOrElse(HBaseMeta.SELECTFIELDS, ""),
      fourMap.getOrElse(HBaseMeta.ROWKEY, "")
    )
  }
}
